package com.sh.data.engine.domain.datadev.flink.job;

import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.FlinkJobStatusEnum;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTypeEnum;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkJobDomain;
import com.sh.data.engine.domain.datadev.flink.service.FlinkJobService;
import com.sh.data.engine.domain.datadev.flinkTaskAboutExexte.FlinkTaskApiService;
import com.sh.data.engine.infrastructure.config.FlinkConfig;
import com.sh.data.engine.job.core.handler.annotation.XxlJob;
import com.sh.data.engine.job.core.util.CmdUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author: zigui.zdf
 * @description:
 * @date: 2021/6/24 13:26
 */
@Component
@Slf4j
public class CheckRunningFlinkTaskStatusTask {

    @Autowired
    private FlinkJobService flinkJobService;

    @Autowired
    private FlinkTaskApiService flinkTaskApiService;

    @Autowired
    private FlinkConfig flinkConfig;

    /**
     * 如果yarn上的application停掉了，或者flink
     * job停掉了，修改虎符的job状态，如果一个job对应多个application，有一个application停了，则所有的都停掉
     */
    @XxlJob("CheckRunningFlinkTaskStatusTask")
    public void checkEngineFlinkTask() {
        final List<FlinkJobDomain> runningJob = flinkJobService.getRunningJobList();

        if (CollectionUtils.isEmpty(runningJob)) {
            return;
        }

        for (FlinkJobDomain flinkJobDomain : runningJob) {
            final Integer taskType = flinkJobDomain.getTaskType();

            if (taskType == FlinkNodeTypeEnum.TASK_SQL.getCode()
                || taskType == FlinkNodeTypeEnum.TASK_JAR.getCode()) {
                this.checkYarnApplication(flinkJobDomain);
            }
        }
    }

    private void checkYarnApplication(FlinkJobDomain flinkJobDomain) {
        final String flinkJobId = flinkJobDomain.getFlinkJobId();

        try {
            Integer taskStatus = flinkJobDomain.getStatus();

            List<String> runningFlinkJobs = getRunningFlinkJobs();

            if (FlinkJobStatusEnum.RUNNING.getCode() != taskStatus
                && StringUtils.isNotBlank(flinkJobDomain.getFlinkJobId())) {
                if (runningFlinkJobs.contains(flinkJobId)) {
                    flinkTaskApiService.stopFlinkTask(flinkJobDomain.getId(), taskStatus, true);
                }
            }

            if (FlinkJobStatusEnum.RUNNING.getCode() == taskStatus
                && StringUtils.isNotBlank(flinkJobDomain.getFlinkJobId())) {
                if (!runningFlinkJobs.contains(flinkJobDomain.getFlinkJobId())) {
                    flinkJobService.updateJobStatus(
                        flinkJobDomain.getId(), FlinkJobStatusEnum.STOPPED.getCode());
                }
            }

        } catch (Exception e) {
            log.error("check running jobId:{} error", flinkJobDomain.getId(), e);
        }
    }

    private List<String> getRunningFlinkJobs() {
        String flinkHome = flinkConfig.getHome();
        String cmd = flinkHome + "flink list";
        Integer startIndex = 0;
        Integer endIndex = 0;
        List<String> flink = Lists.newArrayList();
        List<String> runningStr = Lists.newArrayList();
        List<String> runningJobIds = Lists.newArrayList();
        try {
            // flink = runCmd("root", args.toArray(new String[args.size()]));
            String[] result = CmdUtil.runCmd(cmd);
            flink = Arrays.stream(result).collect(Collectors.toList());
            for (int i = 0; i < flink.size(); i++) {
                String line = flink.get(i);
                int i1 = line.indexOf("------------------ Running/Restarting Jobs -------------------");
                int i2 = line.indexOf("--------------------------------------------------------------");
                if (i1 != -1) {
                    startIndex = i;
                }
                if (i2 != -1) {
                    endIndex = i;
                }
            }
        } catch (Exception e) {

        }
        if (startIndex > 0 && endIndex > startIndex) {
            runningStr = flink.subList(startIndex + 1, endIndex);
        }
        runningStr.forEach(
            i -> {
                String[] split = i.split(":");
                runningJobIds.add(split[3].trim());
            });
        return runningJobIds;
    }
}
